#include "sock_msg.h"
#include <netinet/in.h>
#include <signal.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>

#include <linux/un.h>
#include <sys/socket.h>

#include "msg_def.h"
#include "common.h"
#include "gp_layout_map.h"
#include "defs.grpc.pb.h"
#include "defs_self.grpc.pb.h"
#include "motor_config.h"
#include "spit_param.h"
#include <fcntl.h>
#include "motor_config.h"

int cliFd = -1;
int sock_exitFlag = 0;

typedef struct _ack_msg_node
{
    msgAckSock msgHead;
    void * paramData;
    struct _ack_msg_node * next;
}ack_msg_node;

typedef struct _msg_buff_list
{
    ack_msg_node * head;
    ack_msg_node * tail;
}msg_buff_list;

msg_buff_list g_list_buff_msg;

int init_msg_buf_node(char * dataFromSock, int sockDataLen,ack_msg_node ** nodeOut );
void append_msg_buff_node(ack_msg_node * nodeIn);

int initSockClient(void)
{
    int                 fd;
    struct sockaddr_un  serveraddr;

    printf("enter in initSockClient\n");
 
    fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (fd < 0) {
        fprintf(stderr, "socket: %s\n", strerror(errno));
        return -1;
    }
 
    serveraddr.sun_family = AF_UNIX;
    snprintf(serveraddr.sun_path, sizeof(serveraddr.sun_path), "%s", SOCK_IPC_FILE);
    
    int flags=fcntl(fd,F_GETFL,0);
    fcntl(fd,F_SETFL,flags|O_NONBLOCK);//设置为非阻塞

    int result = connect(fd, (struct sockaddr *)&serveraddr, (socklen_t)sizeof(struct sockaddr_un));
    if (result)
    {
        perror("ops: client\n");
        //exit(1);
        return -1;
    }
    else
    {
        cliFd = fd;
        pthread_t thread_handle;
        pthread_create(&thread_handle,NULL,sockRecvFunc,NULL);
    }
    return 0;
}


void * sockRecvFunc(void * paramIn)
{
    fd_set rfds;
    struct timeval tv;
    int retval;
    tv.tv_sec = 2;
    tv.tv_usec = 0;
    char recvBuff[256] = {0};
   while(!sock_exitFlag)
   {
       FD_ZERO(&rfds);
       if(cliFd<0)
       {
           break;
       }
       FD_SET(cliFd, &rfds);

       retval = select(cliFd+1,&rfds,NULL,NULL,&tv);

       if(retval<0)
       {
           perror("select()");
           close(cliFd);
           cliFd = -1;
       }
       else if(retval==0)
       {
           //no data available in 2 second;
       }
       else
       {
           //data received;
           if(FD_ISSET(cliFd, &rfds))
           {
               //printf("received data in sockRecvFunc,");

               int bytesRead = read(cliFd,recvBuff,256);
               
               if(bytesRead>0)
               {
                   //printf("data len:%d\n",bytesRead);
                   ack_msg_node * pstNode = NULL;
                   init_msg_buf_node(recvBuff,bytesRead,&pstNode);

                   if(pstNode != NULL)
                   {
                       append_msg_buff_node(pstNode);
                   }
               }
               else
               {
                   //printf("data len:%d\n",bytesRead);
                   printf("remote socket closed\n");
                   close(cliFd);
                   cliFd = -1;
               }
           }
       }
   } 

   pthread_detach(pthread_self());

   return (void*)NULL;
}

pthread_mutex_t mutex_sock_send = PTHREAD_MUTEX_INITIALIZER;

int encodeTransPos(int moduleType, sockActMsg * msgHeadIn, void * paramIn, char * buffOut)
{
    GPRPC::JawMoveParam * pstTransParam = (GPRPC::JawMoveParam *)paramIn;
    location_layout locSrc;
    location_layout locDst;

    int convertRet = 0;
    int lenRet = 0;

    char * pBuffOut = buffOut;
    
    if(moduleType == GPRPC::ModuleType::MT_DP1
    || moduleType == GPRPC::ModuleType::MT_DJ1_PCR_REAGENT
    || moduleType == GPRPC::ModuleType::MT_DP8
    || moduleType == GPRPC::ModuleType::MT_LOGISTICS)
    {
        convertRet = convertGRPCPos(moduleType, pstTransParam->pickuppos(),&locSrc);
        if(convertRet)
        {
            return lenRet;
        }

        convertRet = convertGRPCPos(moduleType, pstTransParam->releasepos(),&locDst);

        if(convertRet)
        {
            printf("encodeTransPos, convert pos failed, pos type:%d\n",pstTransParam->releasepos().postype());
            return lenRet;
        }

        msgHeadIn->paramLen = sizeof(location_layout)*2;

        lenRet = encodeMsgHead(msgHeadIn,pBuffOut,0);

        pBuffOut += lenRet;

        memcpy(pBuffOut,&locSrc,sizeof(location_layout));

        pBuffOut +=  sizeof(location_layout);

        memcpy(pBuffOut,&locDst,sizeof(location_layout));

        lenRet += sizeof(location_layout)*2;
    }
    

    return lenRet;
}

int encodeJawCtl(int moduleType, sockActMsg * msgHeadIn, void * paramIn, char * buffOut)
{
    GPRPC::JawGetPutParam * pstTransParam = (GPRPC::JawGetPutParam *)paramIn;
    location_layout locSrc;
    location_layout locDst;

    int convertRet = 0;
    int lenRet = 0;

    char * pBuffOut = buffOut;

    memset(&locSrc,0,sizeof(location_layout));
    memset(&locDst,0,sizeof(location_layout));
    
    if(moduleType == GPRPC::ModuleType::MT_DP1
    || moduleType == GPRPC::ModuleType::MT_DJ1_PCR_REAGENT
    || moduleType == GPRPC::ModuleType::MT_DP8
    || moduleType == GPRPC::ModuleType::MT_LOGISTICS)
    {
        if(pstTransParam->ctrl() == GPRPC::Switch::CLOSE)
        {
            convertRet = convertGRPCPos(moduleType, pstTransParam->pos(),&locSrc);
            if(convertRet)
            {
                return lenRet;
            }        
        }
        else
        {
            locSrc.location_type = 0xffff;
        }

        if(pstTransParam->ctrl() == GPRPC::Switch::OPEN)
        {
            convertRet = convertGRPCPos(moduleType, pstTransParam->pos(),&locDst);

            if(convertRet)
            {
                printf("encodeTransPos, convert pos failed, pos type:%d\n",pstTransParam->pos().postype());
                return lenRet;
            }            
        }
        else
        {
            locDst.location_type = 0xffff;
        }
        
        msgHeadIn->paramLen = sizeof(location_layout)*2;

        lenRet = encodeMsgHead(msgHeadIn,pBuffOut,0);

        pBuffOut += lenRet;

        memcpy(pBuffOut,&locSrc,sizeof(location_layout));

        pBuffOut +=  sizeof(location_layout);

        memcpy(pBuffOut,&locDst,sizeof(location_layout));

        lenRet += sizeof(location_layout)*2;
    }
    

    return lenRet;
}

int encodeFilmTrans(sockActMsg * msgHeadIn, int fileIndex, char * buffOut)
{
    location_layout locSrc;
    location_layout locDst;

    int convertRet = 0;
    int lenRet = 0;

    char * pBuffOut = buffOut; 

    memset(&locSrc,0,sizeof(location_layout));
    memset(&locDst,0,sizeof(location_layout));

    locSrc.location_type = DP8_FILM_SRC; 

    locSrc.indexX = fileIndex;   

    locDst.location_type = DP8_HEAT_SEAL;

    msgHeadIn->paramLen = sizeof(location_layout)*2;

    lenRet = encodeMsgHead(msgHeadIn,pBuffOut,0);

    pBuffOut += lenRet;

    memcpy(pBuffOut,&locSrc,sizeof(location_layout));

    pBuffOut +=  sizeof(location_layout);

    memcpy(pBuffOut,&locDst,sizeof(location_layout));

    lenRet += sizeof(location_layout)*2;
    

    return lenRet;
}

int encodeFilmTrans_en(sockActMsg * msgHeadIn, int fileIndex, char * buffOut, int isPut)
{
    location_layout locSrc;
    location_layout locDst;

    int convertRet = 0;
    int lenRet = 0;

    char * pBuffOut = buffOut; 

    memset(&locSrc,0,sizeof(location_layout));
    memset(&locDst,0,sizeof(location_layout));

    if(isPut)
    {
        locSrc.location_type = 0xffff;  
        locDst.location_type = DP8_HEAT_SEAL;
    }
    else
    {
        locSrc.location_type = DP8_FILM_SRC;
        locSrc.indexX = fileIndex; 
        locDst.location_type = 0xffff;
    }

    msgHeadIn->paramLen = sizeof(location_layout)*2;

    lenRet = encodeMsgHead(msgHeadIn,pBuffOut,0);

    pBuffOut += lenRet;

    memcpy(pBuffOut,&locSrc,sizeof(location_layout));

    pBuffOut +=  sizeof(location_layout);

    memcpy(pBuffOut,&locDst,sizeof(location_layout));

    lenRet += sizeof(location_layout)*2;
    

    return lenRet;
}

int encodeLogMove(sockActMsg * msgHeadIn, const void * paramIn, char * buffOut)
{
    location_layout locDst;

    int convertRet = 0;
    int lenRet = 0;
    logistic_move *  pstParam = (logistic_move *)paramIn;
    


    char * pBuffOut = buffOut;
    memset(&locDst,0,sizeof(location_layout));

    locDst.location_type = pstParam->posTypeInner;
    locDst.indexX = pstParam->submodule;

    msgHeadIn->paramLen = sizeof(location_layout);

    lenRet = encodeMsgHead(msgHeadIn,pBuffOut,0);

    pBuffOut += lenRet;

    memcpy(pBuffOut,&locDst,sizeof(location_layout));

    lenRet += sizeof(location_layout);    

    return lenRet;
}

int encodeLogTrans(sockActMsg * msgHeadIn, const void * paramIn, char * buffOut)
{
    location_layout locDst;
    location_layout locSrc;
    int convertRet = 0;
    int lenRet = 0;
    logistic_move *  pstParam = (logistic_move *)paramIn;
    


    char * pBuffOut = buffOut;
    memset(&locSrc,0,sizeof(location_layout));
    memset(&locDst,0,sizeof(location_layout));

    locSrc.location_type = pstParam->posTypeInner;
    locSrc.indexX = pstParam->submodule;

    pstParam++;

    locDst.location_type = pstParam->posTypeInner;
    locDst.indexX = pstParam->submodule;

    msgHeadIn->paramLen = sizeof(location_layout)*2;

    lenRet = encodeMsgHead(msgHeadIn,pBuffOut,0);

    pBuffOut += lenRet;


    memcpy(pBuffOut,&locSrc,sizeof(location_layout));

    pBuffOut += sizeof(location_layout);

    memcpy(pBuffOut,&locDst,sizeof(location_layout));

    lenRet += sizeof(location_layout)*2;    

    return lenRet;
}


int encodeSuckMsg(sockActMsg * pstMsgHeadIn,int moduleTypeGrpc,const GPRPC::MoveLiquidParam & pstMoveLiquid,int isForMulti, char * buffOut)
{
    pstMsgHeadIn->paramLen = sizeof(location_layout) + (sizeof(int)<<2) + sizeof(int);
    location_layout locLayout;
    memset(&locLayout,0,sizeof(location_layout));

    convertGRPCPos(moduleTypeGrpc, pstMoveLiquid.pos(),&locLayout);
    char * buffSend = buffOut;

    int currOffset = encodeMsgHead(pstMsgHeadIn,buffSend); 
    int liquidType = pstMoveLiquid.viscosity();//-----
    int suckVolume = pstMoveLiquid.volum();
    int remainVolume = pstMoveLiquid.existv();
    int tipType = pstMoveLiquid.tiptype();
    //printf("encodeSuckMsg, tipType is %d\n",tipType);
    memcpy(buffSend+currOffset,&locLayout,sizeof(location_layout));     
    currOffset += sizeof(location_layout);
    //|xylayout|liquidtype|suckVolume|isformulti|remainVolume
    currOffset += encodeInt(liquidType,buffSend+currOffset);
    currOffset += encodeInt(suckVolume,buffSend+currOffset);
    currOffset += encodeInt(isForMulti,buffSend+currOffset);
    currOffset += encodeInt(remainVolume,buffSend+currOffset);
    currOffset += encodeInt(tipType,buffSend+currOffset);
    printf("suck msg, remain:%d,tiptype:%d\n",remainVolume,tipType);
    return currOffset;
}

int encodeSpitMsg_multi(sockActMsg * pstMsgHeadIn,pos_one_type * spitParamIn, char * buffOut,int isformultiIn, int remainVolume)
{
    pstMsgHeadIn->paramLen = sizeof(location_layout) + (sizeof(int)<<1) + (sizeof(int)<<1);
    location_layout locLayout;
    memset(&locLayout,0,sizeof(location_layout));

    int moduleTypeGrpc = pstMsgHeadIn->moduleType;

    GPRPC::Position posIn;
    posIn.set_column(spitParamIn->posListHead->column_x);
    posIn.set_row(spitParamIn->posListHead->row_y);
    posIn.set_index(GPRPC::Index(spitParamIn->posListHead->index) );
    posIn.set_postype(GPRPC::PosType(spitParamIn->posType));

    convertGRPCPos(moduleTypeGrpc, posIn,&locLayout);
    char * buffSend = buffOut;
//|xylayout|spitVolume|isformulti|remainVolume
    int currOffset = encodeMsgHead(pstMsgHeadIn,buffSend); 
    int headLen = currOffset;
    int liquidType = VISCOSITY_LOW;//-----
    memcpy(buffSend+currOffset,&locLayout,sizeof(location_layout));     
    currOffset += sizeof(location_layout);
    //currOffset += encodeInt(liquidType,buffSend+currOffset);
    currOffset += encodeInt(spitParamIn->posListHead->volume,buffSend+currOffset);
    currOffset += encodeInt(isformultiIn,buffSend+currOffset);
    //int remainVolume = 0;
    currOffset += encodeInt(remainVolume,buffSend+currOffset);
    currOffset += encodeInt(spitParamIn->tipType,buffSend+currOffset);

    std::cout<<"in encodeSpitMsg_multi,remain = "<<remainVolume<<std::endl;

    
    int paramOnce = 0;

    pos_spit * pstPosSpit = spitParamIn->posListHead->next;
    location_layout locLayoutTmp;
    char deltax = 0;
    char deltay = 0;
    unsigned short volume1 = 0;
    char * ptrTempData = NULL;
    while(pstPosSpit)
    {
        posIn.set_column(pstPosSpit->column_x);
        posIn.set_row(pstPosSpit->row_y);
        posIn.set_index(GPRPC::Index(pstPosSpit->index) );
        posIn.set_postype(GPRPC::PosType(spitParamIn->posType));

        convertGRPCPos(moduleTypeGrpc, posIn,&locLayoutTmp);
        deltax = locLayoutTmp.column_x - locLayout.column_x;
        deltay = locLayoutTmp.row_y - locLayout.row_y;
        volume1 = spitParamIn->posListHead->volume;
        // paramOnce = locLayoutTmp.column_x - locLayout.column_x;
        // paramOnce += ((locLayoutTmp.row_y - locLayout.row_y) << 4);
        // paramOnce += ((spitParamIn->posListHead->volume) << 8);
        // paramOnce = 0;
        // ptrTempData = (char *)&paramOnce;
        // *ptrTempData = deltax;
        // *(ptrTempData+1) = deltay;
        // memcpy(ptrTempData+2,)
        memcpy(buffSend+currOffset,&deltax,sizeof(char));
        currOffset+=sizeof(char);
        memcpy(buffSend+currOffset,&deltay,sizeof(char));
        currOffset+=sizeof(char);
        memcpy(buffSend+currOffset,&volume1,sizeof(short));
        currOffset+=sizeof(short);
        
        pstPosSpit = pstPosSpit->next;

        //currOffset += encodeInt(paramOnce,buffSend+currOffset);
    }
    //adjust the paramlen in header.
    pstMsgHeadIn->paramLen = currOffset - headLen;
    pstMsgHeadIn->moduleType = moduleTypeGrpc;
    encodeMsgHead(pstMsgHeadIn,buffSend); 
    
    return currOffset;
}

int encodeSpitData_multi(sockActMsg * pstMsgHeadIn,pos_one_type * spitParamIn, char * buffOut,int isformultiIn, int remainVolume)
{
    //pstMsgHeadIn->paramLen = sizeof(location_layout) + (sizeof(int)<<1) + sizeof(int);
    location_layout locLayout;
    memset(&locLayout,0,sizeof(location_layout));

    int moduleTypeGrpc = pstMsgHeadIn->moduleType;

    GPRPC::Position posIn;
    posIn.set_column(spitParamIn->posListHead->column_x);
    posIn.set_row(spitParamIn->posListHead->row_y);
    posIn.set_index(GPRPC::Index(spitParamIn->posListHead->index) );
    posIn.set_postype(GPRPC::PosType(spitParamIn->posType));

    convertGRPCPos(moduleTypeGrpc, posIn,&locLayout);
    char * buffSend = buffOut;
//|xylayout|spitVolume|isformulti|remainVolume
    int currOffset = 0;//encodeMsgHead(pstMsgHeadIn,buffSend); 

    memcpy(buffSend,&locLayout,sizeof(location_layout));     
    currOffset += sizeof(location_layout);
    //currOffset += encodeInt(liquidType,buffSend+currOffset);
    currOffset += encodeInt(spitParamIn->posListHead->volume,buffSend+currOffset);
    currOffset += encodeInt(isformultiIn,buffSend+currOffset);
    //int remainVolume = 0;
    currOffset += encodeInt(remainVolume,buffSend+currOffset);

    currOffset += encodeInt(spitParamIn->tipType,buffSend+currOffset);
    int paramOnce = 0;

    pos_spit * pstPosSpit = spitParamIn->posListHead->next;
    location_layout locLayoutTmp;
    char deltax = 0;
    char deltay = 0;
    unsigned short volume1 = 0;
    char * ptrTempData = NULL;
    int numTemp = spitParamIn->num;
    while(pstPosSpit && numTemp>=1)
    {
        posIn.set_column(pstPosSpit->column_x);
        posIn.set_row(pstPosSpit->row_y);
        posIn.set_index(GPRPC::Index(pstPosSpit->index) );
        posIn.set_postype(GPRPC::PosType(spitParamIn->posType));

        convertGRPCPos(moduleTypeGrpc, posIn,&locLayoutTmp);
        deltax = locLayoutTmp.column_x - locLayout.column_x;
        deltay = locLayoutTmp.row_y - locLayout.row_y;
        volume1 = spitParamIn->posListHead->volume;

        memcpy(buffSend+currOffset,&deltax,sizeof(char));
        currOffset+=sizeof(char);
        memcpy(buffSend+currOffset,&deltay,sizeof(char));
        currOffset+=sizeof(char);
        memcpy(buffSend+currOffset,&volume1,sizeof(short));
        currOffset+=sizeof(short);
        
        pstPosSpit = pstPosSpit->next;
        numTemp--;
    }
    
    return currOffset;
}

int sendSockMsgToAct(void * msgHead, const void * paramIn)
{
    int errRet = 0;
    
    sockActMsg * pstMsgHead = (sockActMsg *)msgHead;

    int currOffset = 0;

    char buffSend[512];
    char * ptrBuff;
    memset(buffSend,0,512);

    int lenMsg = 0;

    if(pstMsgHead->paramType == PARAM_AXIS_MOVE_POS)
    {
        location_layout locLayout;
        memset(&locLayout,0,sizeof(location_layout));
        pstMsgHead->paramLen = sizeof(location_layout) + sizeof(int);
        int axisType = -1;
        if(pstMsgHead->actType == ACT_SINGLE_MOVE)
        {
            if(pstMsgHead->moduleType == GPRPC::ModuleType::MT_QPCR1
               || pstMsgHead->moduleType == GPRPC::ModuleType::MT_QPCR2)
            {
                qpcr_move * pstQpcrMove = (qpcr_move *)paramIn;
                axisType = pstQpcrMove->motorType;
                locLayout.location_type = pstQpcrMove->pos;
                std::cout<<"QPCR single move pos,posType:"<<axisType<<std::endl;
            }
            else
            {
                GPRPC::MotorPosParam * pstMotorPos = (GPRPC::MotorPosParam *)paramIn;
                convertGRPCPos(pstMotorPos->moduletype(), pstMotorPos->pos(),&locLayout);
                axisType = pstMotorPos->motortype();
            }
        }
        else if(pstMsgHead->actType == ACT_PARRAL_MOVE)
        {
            GPRPC::ModuleMoveParam * pstModulePos = (GPRPC::ModuleMoveParam *)paramIn;
            int convRet = convertGRPCPos(pstModulePos->moduletype(), pstModulePos->pos(),&locLayout);
            if(convRet)
            {
                std::cout<<"ACT_PARRAL_MOVE, convertGRPCPos failed"<<std::endl;
                return 4;
            }

            std::cout<<"ACT_PARRAL_MOVE, loctype:"<<locLayout.location_type <<std::endl;
        }
        else if(pstMsgHead->actType == ACT_SINGLE_INIT)
        {
            GPRPC::Motor * pstMotorParam = (GPRPC::Motor *)paramIn;
            axisType = pstMotorParam->motortype();
        }
        else if(pstMsgHead->actType == ACT_MODULE_INIT)
        {
            
        }
        else if(ACT_PLUNGER_PUMP == pstMsgHead->actType)
        {
            GPRPC::PlungerPumpCtrlParam * pstParam = (GPRPC::PlungerPumpCtrlParam *)paramIn;
            locLayout.column_x = pstParam->volum();

            if(pstParam->pumpctrl() == GPRPC::PumpCtrl::SPIT)
            {
                locLayout.column_x = 0 - locLayout.column_x;
            }

            if(pstParam->pumptype() == GPRPC::PumpType::DJ2_PUMP_SMALL)
            {
                axisType = MT_PUMP_SMALL;
            }
            else
            {
                axisType = MT_PUMP;
            }
        }
        currOffset = encodeMsgHead(pstMsgHead,buffSend);      

        currOffset += encodeInt(axisType,buffSend+currOffset);

        printf("location_type:%d,mx:%d,my:%d,ix:%d,iy:%d\n",
        locLayout.location_type,
        locLayout.indexX,
        locLayout.indexY,
        locLayout.column_x,
        locLayout.row_y);

        memcpy(buffSend+currOffset,&locLayout,sizeof(location_layout));

        lenMsg = currOffset + sizeof(location_layout);
        
    }
    else if(PARAM_AXIS_MOVE_STEPS == pstMsgHead->paramType)
    {
        pstMsgHead->paramLen = sizeof(int) * 2;
        currOffset = encodeMsgHead(pstMsgHead,buffSend);
        GPRPC::MotorMoveParam * pstGrpcMoveParam = (GPRPC::MotorMoveParam *)paramIn;
        
        int moveComponent = pstGrpcMoveParam->motortype();

        int stepsToMove = pstGrpcMoveParam->step();

        memcpy(buffSend+currOffset,&moveComponent,sizeof(int));

        currOffset+= sizeof(int);

        memcpy(buffSend+currOffset,&stepsToMove,sizeof(int));

        currOffset+= sizeof(int);

        lenMsg = currOffset;
    }
    else if(PARAM_LIQUID_CTL == pstMsgHead->paramType)
    {
        GPRPC::DjDpLiquidCtrlParam * ptrLiquidParam = (GPRPC::DjDpLiquidCtrlParam *)paramIn;
        
        if(pstMsgHead->actType == ACT_SUCK_LIQUID)
        {
            int posSize = ptrLiquidParam->moveliquidparam_size();
            if(posSize<1)
            {
                errRet = 3;
                return  errRet;
            }
          
            lenMsg = encodeSuckMsg(pstMsgHead,ptrLiquidParam->moduletype(),ptrLiquidParam->moveliquidparam(0),0, buffSend);
        }
        else if(ACT_SPIT_LIQUID == pstMsgHead->actType)
        {
           
            pos_one_type * pstOneSpitParam = (pos_one_type *)paramIn;

            lenMsg = encodeSpitMsg_multi(pstMsgHead,pstOneSpitParam,buffSend,0,pstOneSpitParam->remainVolume);
            printf("encode msg for spit\n");         

        }        
        //GPRPC::ModuleMoveParam * pstModulePos = (GPRPC::ModuleMoveParam *)paramIn;
        
        
    }
    else if(PARAM_GET_TIP == pstMsgHead->paramType 
         || PARAM_DROP_TIP == pstMsgHead->paramType)
    {
        GPRPC::DpTipCtrlParam * pstTipCtlParam = (GPRPC::DpTipCtrlParam *)paramIn;
        location_layout locLayout;
        convertGRPCPos(pstTipCtlParam->moduletype(), pstTipCtlParam->pos(),&locLayout);
        pstMsgHead->paramLen = sizeof(location_layout) + sizeof(int);

        int tipType = pstTipCtlParam->tiptype();//
        currOffset = encodeMsgHead(pstMsgHead,buffSend);

        printf("location_type:%d,mx:%d,my:%d,ix:%d,iy:%d\n",
        locLayout.location_type,locLayout.indexX,locLayout.indexY,
        locLayout.indexX,locLayout.indexY);

        memcpy(buffSend+currOffset,&locLayout,sizeof(location_layout));      
        currOffset += sizeof(location_layout);
        currOffset += encodeInt(tipType,buffSend+currOffset);
        
        lenMsg = currOffset;
    }
    else if(pstMsgHead->paramType == PARAM_SUCK1_SPIT1)
    {
        if(pstMsgHead->actType == ACT_SUCK_LIQUID)
        {
            GPRPC::DjDpSuckASpitParam * pstSuckParam = (GPRPC::DjDpSuckASpitParam *)paramIn;
            int liquidType = pstSuckParam->pickupparam().viscosity();
            int isForMulti = 0;
            if(pstSuckParam->moduletype() == GPRPC::ModuleType::MT_DP8
                &&pstSuckParam->pickupparam().pos().postype() == GPRPC::PosType::PT_DP1_DP8_SHAKE)
            {
                isForMulti = 1;
            }
            lenMsg = encodeSuckMsg(pstMsgHead,pstSuckParam->moduletype(),pstSuckParam->pickupparam(),isForMulti,buffSend);
            
        }
        else if(pstMsgHead->actType == ACT_SPIT_LIQUID)
        {
            pos_one_type * pstOneSpitParam = (pos_one_type *)paramIn;

            lenMsg = encodeSpitMsg_multi(pstMsgHead,pstOneSpitParam,buffSend,0,pstOneSpitParam->remainVolume);
            printf("encode msg for suck1 spit1, spit\n");
        }
    }
    else if(pstMsgHead->paramType == PARAM_SUCK1_SPIT_MULTI)
    {
        if(pstMsgHead->actType == ACT_SUCK_LIQUID)
        {
            GPRPC::DjDpSuckMoreSpitParam * pstSuckParam = (GPRPC::DjDpSuckMoreSpitParam *)paramIn;
            
            lenMsg = encodeSuckMsg(pstMsgHead,pstSuckParam->moduletype(),pstSuckParam->pickupparam(),1, buffSend);
            
            printf("set isForMulti to 1,lenMsg:%d\n",lenMsg);
        }
        else if(pstMsgHead->actType == ACT_SPIT_LIQUID_MULTI)
        {
            pos_one_type * pstOneSpitParam = (pos_one_type *)paramIn;

            lenMsg = encodeSpitMsg_multi(pstMsgHead,pstOneSpitParam,buffSend,1,pstOneSpitParam->remainVolume);
            printf("encode msg for suck1 spit n, spit\n");
        }
        else if(pstMsgHead->actType == ACT_SUCK1_SPITM)
        {
            param_suck1_spitm * pstSuck1SpitM = (param_suck1_spitm *)paramIn;
            
            int moduleTypeGrpc = pstMsgHead->moduleType;
            currOffset = encodeMsgHead(pstMsgHead,buffSend);
            int dataLen = 0;
            encodeSuck1SpitMData(moduleTypeGrpc, pstSuck1SpitM->stSuckParam, pstSuck1SpitM->pSpitMOnce,
             buffSend + currOffset, &dataLen);

            pstMsgHead->paramLen = dataLen;

            encodeMsgHead(pstMsgHead,buffSend);

            lenMsg = currOffset + dataLen; 
        }
    }
    else if(PARAM_JAW_TRANS == pstMsgHead->paramType)
    {
        GPRPC::JawMoveParam * pstTransPos = (GPRPC::JawMoveParam *)paramIn;

        lenMsg = encodeTransPos(pstTransPos->moduletype(),pstMsgHead,pstTransPos,buffSend);
        printf("encode msg for jaw trans\n");
    }
    else if(PARAM_JAW_TRANS_FILM == pstMsgHead->paramType)
    {
        GPRPC::GetFilmParam * pstFilmParam = (GPRPC::GetFilmParam *)paramIn;
        pstMsgHead->paramType = PARAM_JAW_TRANS;
        lenMsg = encodeFilmTrans(pstMsgHead,pstFilmParam->getpos(),buffSend);
        printf("encode msg for jaw trans film\n");
    }
    else if(PARAM_JAW_CTL== pstMsgHead->paramType)
    {
        GPRPC::JawGetPutParam * pstJawCtl = (GPRPC::JawGetPutParam *)paramIn;
        pstMsgHead->paramType = PARAM_JAW_TRANS;
        lenMsg = encodeJawCtl(pstJawCtl->moduletype(),pstMsgHead,pstJawCtl,buffSend);
        printf("encode msg for jaw ctl\n");
    }
    else if(PARAM_JAW_CTL_FILM== pstMsgHead->paramType)
    {
        GPRPC::GetPutFilmParam * pstJawCtl = (GPRPC::GetPutFilmParam *)paramIn;
        pstMsgHead->paramType = PARAM_JAW_TRANS;
        int isPut = 0;
        if(pstJawCtl->switch_() == GPRPC::Switch::OPEN)
        {
            isPut = 1;
        }
        else
        {
            isPut = 0;
        }
        lenMsg = encodeFilmTrans_en(pstMsgHead,pstJawCtl->getpos(),buffSend,isPut);
        printf("encode msg for film getput\n");
    }
    else if(PARAM_LOGISTIC_MOVE== pstMsgHead->paramType)
    {
        if(ACT_PARRAL_MOVE == pstMsgHead->actType)
        {
            logistic_move * pstLogMove = (logistic_move *)paramIn;
            lenMsg = encodeLogMove(pstMsgHead,pstLogMove,buffSend);
            printf("encode msg for logistic move\n");
        }
        else if(ACT_TRANS_BOARD == pstMsgHead->actType)
        {
            logistic_move * pstLogMove = (logistic_move *)paramIn;
            lenMsg = encodeLogTrans(pstMsgHead,pstLogMove,buffSend);
            printf("encode msg for logistic trans\n");            
        }

    }

    if(lenMsg)
    {
        if(cliFd==-1)
        {
            initSockClient();
        }

        if(cliFd==-1)
        {
            errRet = 2;
            printf("msg connection break down\n");
            return 1;
        }
        pthread_mutex_lock(&mutex_sock_send);
        int bytesWrite = write(cliFd, buffSend, lenMsg);
        pthread_mutex_unlock(&mutex_sock_send);
        if(bytesWrite < 0){
            printf("msg send failed,error:%d\n",errno);
        }
        else 
        {
            printf("msg send success,len:%d\n",bytesWrite);
        }
    }
    else
    {
        errRet = 1;
    }    

    return errRet;
}

int encode_dj_clean_ctl(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    GPRPC::DjPumpValveSwitchParam * pvCtlParam = (GPRPC::DjPumpValveSwitchParam *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = 1;
    currOffset = encodeMsgHead(pstMsgHead,buffOut);
    int isStart = 0;
    
    if(pvCtlParam->switch_() == GPRPC::Switch::OPEN)
    {
        isStart = 1;
    }
    else
    {
        isStart = 0;
    }
    memcpy(buffOut+currOffset,&isStart,sizeof(int));
    currOffset += sizeof(int);
    return currOffset;
}

int encode_shake_ctl(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    switch_ctl * pvCtlParam = (switch_ctl *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = 2*sizeof(int);
    pstMsgHead->paramType = pvCtlParam->chanNum;
    std::cout<<"in encode_shake_ctl, paramType:"<<pstMsgHead->paramType<<std::endl;
    currOffset = encodeMsgHead(pstMsgHead,buffOut,1);
    int isStart = 0;
    
    if(pvCtlParam->isClose)
    {
        isStart = 0;
    }
    else
    {
        isStart = 1;
    }
    memcpy(buffOut+currOffset,&isStart,sizeof(int));
    currOffset += sizeof(int);

    int duration = pvCtlParam->duration;

    memcpy(buffOut+currOffset,&duration,sizeof(int));
    currOffset += sizeof(int);

    return currOffset;
}

int encode_light_ctl(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    switch_ctl * pvCtlParam = (switch_ctl *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = 2*sizeof(int);
    pstMsgHead->paramType = pvCtlParam->chanNum;
    std::cout<<"in encode_shake_ctl, paramType:"<<pstMsgHead->paramType<<std::endl;
    currOffset = encodeMsgHead(pstMsgHead,buffOut,1);
    int isStart = 0;
    
    if(pvCtlParam->isClose)
    {
        isStart = 0;
    }
    else
    {
        isStart = 1;
    }
    memcpy(buffOut+currOffset,&isStart,sizeof(int));
    currOffset += sizeof(int);

    int duration = pvCtlParam->duration;

    memcpy(buffOut+currOffset,&duration,sizeof(int));
    currOffset += sizeof(int);

    return currOffset;
}

int encode_heat_ctl(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    heat_ctl * pvCtlParam = (heat_ctl *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = 2*sizeof(int);
    pstMsgHead->paramType = pvCtlParam->chanNum;
    currOffset = encodeMsgHead(pstMsgHead,buffOut,1);
    int isStart = 0;
    
    if(pvCtlParam->isClose)
    {
        isStart = 0;
    }
    else
    {
        isStart = 1;
    }
    memcpy(buffOut+currOffset,&isStart,sizeof(int));
    currOffset += sizeof(int);

    float temperature = pvCtlParam->temperature;
    printf("temperature:%f\n",pvCtlParam->temperature);

    memcpy(buffOut+currOffset,&temperature,sizeof(float));
    currOffset += sizeof(float);

    return currOffset;
}


int encode_chain(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;

	if(pstMsgHead->moduleType == MODULE_OPENLID)
	{
	    const GPRPC::SwitchCtrl* pstSw = (const GPRPC::SwitchCtrl*)paramIn;

		int isStart = 0;
		
		if(pstSw->switch_() == GPRPC::Switch::OPEN)
		{
		    isStart = 1;
		}
		else
		{
            isStart = 0;		   
		}

		pstMsgHead->paramLen = sizeof(int);
        currOffset = encodeMsgHead(pstMsgHead,buffOut,1);

		memcpy(buffOut+currOffset,&isStart,sizeof(int));
        currOffset += sizeof(int);
	}
	else if(pstMsgHead->moduleType == MODULE_SEAL)
	{
		pstMsgHead->paramLen = sizeof(int);

        const int * pSealHeight = (const int *)paramIn;

        int downSteps = *pSealHeight;
        currOffset = encodeMsgHead(pstMsgHead,buffOut,1);

		memcpy(buffOut+currOffset,&downSteps,sizeof(int));
        currOffset += sizeof(int);
	}
	
    return currOffset;
}

int encode_scheme_ctl(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    switch_ctl * pvCtlParam = (switch_ctl *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = sizeof(int);
    pstMsgHead->paramType = 0;
    currOffset = encodeMsgHead(pstMsgHead,buffOut,1);
    int isStart = 0;
    
    if(pvCtlParam->isClose)
    {
        isStart = 0;
    }
    else
    {
        isStart = 1;
    }
    memcpy(buffOut+currOffset,&isStart,sizeof(int));
    currOffset += sizeof(int);

    return currOffset;
}


int encode_temperature_read(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    //GPRPC::DjPumpValveSwitchParam * pvCtlParam = (GPRPC::DjPumpValveSwitchParam *)paramIn;
    GPRPC::GetTempParam * getTempParam = (GPRPC::GetTempParam *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = sizeof(int)*2;
    currOffset = encodeMsgHead(pstMsgHead,buffOut,1);

    int isStart = (getTempParam->switch_() == GPRPC::CLOSE)?0:1;

    memcpy(buffOut+currOffset,&isStart,4);    
    currOffset += sizeof(int);

    int interval = getTempParam->intervaltime();
    memcpy(buffOut+currOffset,&interval,4);    
    currOffset += sizeof(int);

    return currOffset;
}

int encode_drawer_ctl(const void * paramIn, void * msgHeadInOut,char * buffOut)
{
    GPRPC::DrawerMagLockParam * drawerLockCtlParam = (GPRPC::DrawerMagLockParam *)paramIn;
    sockActMsg * pstMsgHead = (sockActMsg *)msgHeadInOut;
    int currOffset = 0;
    pstMsgHead->paramLen = 1;
    currOffset = encodeMsgHead(pstMsgHead,buffOut,1);
     int isToLock = 0;
    
    if(drawerLockCtlParam->lockstatus() == GPRPC::MagnetLockStatus::MAGNET_LOCK)
    {
        isToLock = 1;
    }
    else
    {
        isToLock = 0;
    }
    memcpy(buffOut+currOffset,&isToLock,sizeof(int));
    currOffset += sizeof(int);
    return currOffset;
}

int sendSockMsgToCtl(void * msgHead, const void * paramIn)
{
    int errRet = 0;
    
    sockActMsg * pstMsgHead = (sockActMsg *)msgHead;

    int currOffset = 0;

    char buffSend[256];
    char * ptrBuff;
    memset(buffSend,0,256);

    int lenMsg = 0;

    if(pstMsgHead->actType == ACT_SCANCODE_CTL)
    {
        printf("encode msg for scancode\n");
        GPRPC::ScancodeParam * pstScanCodeParam = (GPRPC::ScancodeParam *)paramIn;
        pstMsgHead->paramLen = sizeof(int)*2;
        currOffset = encodeMsgHead(pstMsgHead,buffSend,1);
        int chanNum = pstScanCodeParam->scanid();
        int isStart = pstScanCodeParam->isstart();
        memcpy(buffSend+currOffset,&chanNum,sizeof(int));
        currOffset += sizeof(int);
        memcpy(buffSend+currOffset,&isStart,sizeof(int));
        currOffset += sizeof(int);

        lenMsg = currOffset;

    }
    else if(ACT_DRAWER_CTL == pstMsgHead->actType)
    {
        printf("encode msg for drawer ctl\n");
        lenMsg = encode_drawer_ctl(paramIn, pstMsgHead,buffSend);
    }
    else if(ACT_DJ_CLEAN_CTL == pstMsgHead->actType)
    {
        printf("encode msg for dj clean ctl\n");
        lenMsg = encode_dj_clean_ctl(paramIn, pstMsgHead,buffSend);
    }
    else if(ACT_TEMPERATURE_READ == pstMsgHead->actType)
    {
        printf("encode msg for read temperature ctl\n");
        lenMsg = encode_temperature_read(paramIn, pstMsgHead,buffSend);        
    }
    else if(ACT_SHAKE_CTL == pstMsgHead->actType)
    {
        printf("encode msg for shake ctl\n");
        lenMsg = encode_shake_ctl(paramIn, pstMsgHead,buffSend);        
    }
    else if(ACT_HEAT_CTL == pstMsgHead->actType)
    {
        lenMsg = encode_heat_ctl(paramIn, pstMsgHead,buffSend);
    }
	else if(ACT_CHAIN_ACTION == pstMsgHead->actType)
    {
        lenMsg = encode_chain(paramIn, pstMsgHead,buffSend);
    }
    else if(ACT_SCHEME_CTL == pstMsgHead->actType)
    {
        lenMsg = encode_scheme_ctl(paramIn, pstMsgHead,buffSend);
    }
    else if(ACT_LIGHT_CTL == pstMsgHead->actType)
    {
        lenMsg = encode_light_ctl(paramIn, pstMsgHead,buffSend);
    }
    
    if(lenMsg)
    {
        if(cliFd==-1)
        {
            initSockClient();
        }

        if(cliFd==-1)
        {
            errRet = 2;
            printf("msg connection break down\n");
            return errRet;
        }
        pthread_mutex_lock(&mutex_sock_send);
        int bytesWrite = write(cliFd, buffSend, lenMsg);
        pthread_mutex_unlock(&mutex_sock_send);
        if(bytesWrite < 0){
            printf("msg send failed,error:%d\n",errno);
        }
        else
        {
            printf("msg send success, len:%d\n",bytesWrite);
        }
    }
    else
    {
        printf("msg len is 0\n");
        errRet = 1;
    }    

    return errRet;
}

void sendToActAndWaitResp(void * msgHead, const void * paramIn,
int timeMs,act_ctl_node * nodeToCheck,
void * respOut)
{   
    sockActMsg * pstMsgHead = (sockActMsg *)msgHead;
    int sendRet = 0;
    if(  pstMsgHead->actType == ACT_SCANCODE_CTL
      || pstMsgHead->actType == ACT_DJ_CLEAN_CTL
      || pstMsgHead->actType == ACT_DRAWER_CTL
      || pstMsgHead->actType == ACT_TEMPERATURE_READ
      || pstMsgHead->actType == ACT_SHAKE_CTL
      || pstMsgHead->actType == ACT_HEAT_CTL
      || pstMsgHead->actType == ACT_CHAIN_ACTION
      || ACT_SCHEME_CTL == pstMsgHead->actType
      || ACT_LIGHT_CTL == pstMsgHead->actType)
    {
        sendRet = sendSockMsgToCtl(msgHead, paramIn);
    }
    else
    {
        sendRet = sendSockMsgToAct(msgHead, paramIn);
    }
    
    GPRPC::CmdResponse * pstResp = (GPRPC::CmdResponse *)respOut;
    if(sendRet)
    {
        printf("error code:%d\n",sendRet);
        nodeToCheck->status = LUA_ACT_ERROR;
        pstResp->set_returnval(GPRPC::ReturnVal::RET_FALSE);
        return;
    }
    int countMs = timeMs;
    while(countMs>0)
    {
        if(nodeToCheck->status != LUA_ACT_WAITING)
        {
            break;
        }
        usleep(1000);
        countMs--; 
    }

    if(nodeToCheck->status == LUA_ACT_FINISHED)
    {
        pstResp->set_returnval(GPRPC::ReturnVal::RET_TRUE);
    }
    else if(  nodeToCheck->status == LUA_ACT_ERROR 
           || LUA_ACT_TIMEOUT == nodeToCheck->status)
    {
        pstResp->set_returnval(GPRPC::ReturnVal::RET_FALSE);
        std::cout<<"set return value to false"<<std::endl;
    }
    else if(nodeToCheck->status == LUA_ACT_WAITING)
    {
        pstResp->set_returnval(GPRPC::ReturnVal::RET_TIMEOUT);
        nodeToCheck->status = LUA_ACT_TIMEOUT;
    }

}

int encodeMsgHead(void * pstMsgHeadIn, char * buff, int moduleConverted)
{ 
    char * ptrBuff = buff;
    sockActMsg * pstMsgHead = (sockActMsg *)pstMsgHeadIn;
    if(!moduleConverted)
    {
        pstMsgHead->moduleType = convertModule(pstMsgHead->moduleType);
    }

    ptrBuff += encodeInt(pstMsgHead->moduleType,ptrBuff);
    ptrBuff += encodeInt(pstMsgHead->actType,ptrBuff);
    ptrBuff += encodeInt(pstMsgHead->paramType,ptrBuff);
    ptrBuff += encodeInt(pstMsgHead->paramLen,ptrBuff);

    return 4*sizeof(int);

}

int encodeInt(int dataIn,char * buff)
{
    int lenInt = sizeof(int);
    memcpy(buff,&dataIn,lenInt);

    return lenInt;
}

typedef struct _sock_ctl_hash_table
{
   act_ctl_node * table[HASH_SIZE_SOCK_ACT_CTL];
}sock_ctl_hash_table;

sock_ctl_hash_table g_sockActCtlTable;
pthread_mutex_t mutex_table = PTHREAD_MUTEX_INITIALIZER;

void initCtlTable(void)
{
    memset(&g_sockActCtlTable,0,sizeof(act_ctl_node *)*HASH_SIZE_SOCK_ACT_CTL);
}

void clearCtlTable(void)
{
    for(int i=0;i<HASH_SIZE_SOCK_ACT_CTL;i++)
    {
        act_ctl_node *ptrNode = g_sockActCtlTable.table[i];
        act_ctl_node *ptrNodePrev = ptrNode;
        while(ptrNode)
        {
            ptrNodePrev = ptrNode;

            ptrNode = ptrNode->next;
            free(ptrNodePrev);
        }
    }
    memset(&g_sockActCtlTable,0,sizeof(act_ctl_node *)*HASH_SIZE_SOCK_ACT_CTL);
}

act_ctl_node * registerCtlData(sock_act_ctl * dataIn)
{
    int hashCode = (dataIn->actId)%HASH_SIZE_SOCK_ACT_CTL;

    act_ctl_node * ptrCtlDataInHashTable = g_sockActCtlTable.table[hashCode];
    act_ctl_node * pstRet = NULL;
    //printf("enter in registerCheckData, actid[%d]\n",dataIn->actId);
    if(ptrCtlDataInHashTable==NULL)
    {
        printf("registerCtlData, node is null, malloc here\n");
        pthread_mutex_lock(&mutex_table);   
        g_sockActCtlTable.table[hashCode] = (act_ctl_node *)malloc(sizeof(act_ctl_node));
        memset(g_sockActCtlTable.table[hashCode],0,sizeof(act_ctl_node));        
        pthread_mutex_unlock(&mutex_table);

        g_sockActCtlTable.table[hashCode]->actId = dataIn->actId;
        g_sockActCtlTable.table[hashCode]->status = dataIn->status;
        g_sockActCtlTable.table[hashCode]->ptrBack = dataIn;

        pstRet = g_sockActCtlTable.table[hashCode];
    }
    else {

        //find the node with sid == checkDataIn->valueDesired.sid
        act_ctl_node * ptrPre = ptrCtlDataInHashTable;
        while(ptrCtlDataInHashTable)
        {
            if(ptrCtlDataInHashTable->actId == dataIn->actId)
            {
                break;
            }
            ptrPre = ptrCtlDataInHashTable;
            ptrCtlDataInHashTable = ptrCtlDataInHashTable->next;

        }

        //pthread_mutex_lock(&mutex_hash);
        if(ptrCtlDataInHashTable == NULL)
        {//not find
            printf("registerCheckData, node not find, malloc here\n");

            pthread_mutex_lock(&mutex_table);   
            ptrCtlDataInHashTable = (act_ctl_node *)malloc(sizeof(act_ctl_node));
            memset(ptrCtlDataInHashTable,0,sizeof(act_ctl_node));        
            

            ptrCtlDataInHashTable->actId = dataIn->actId;
            ptrCtlDataInHashTable->status = dataIn->status;
            ptrCtlDataInHashTable->ptrBack = dataIn;
            ptrPre->next = ptrCtlDataInHashTable;
            pthread_mutex_unlock(&mutex_table);
        }
        else
        {//find            
            //ptrCtlDataInHashTable->actId = dataIn->actId;
            ptrCtlDataInHashTable->status = dataIn->status;
            ptrCtlDataInHashTable->ptrBack = dataIn;
        }

        pstRet = ptrCtlDataInHashTable;
        //pthread_mutex_unlock(&mutex_hash);
    }

    return pstRet; 
}

act_ctl_node * findCtlNodeInTable(int actId)
{
    int hashCode = actId%HASH_SIZE_SOCK_ACT_CTL;

    act_ctl_node * ptrCtlDataInHashTable = g_sockActCtlTable.table[hashCode];
    act_ctl_node * ptrRet = NULL;
    
    while(ptrCtlDataInHashTable)
    {
        if(ptrCtlDataInHashTable->actId == actId)
        {
            ptrRet = ptrCtlDataInHashTable;
            break;
        }
        ptrCtlDataInHashTable = ptrCtlDataInHashTable->next;
    }
    
    
    return ptrRet;
}

int updateCtlNode(int actIdIn,int statusIn, void * dataIn)
{
    act_ctl_node *nodeResult =  findCtlNodeInTable(actIdIn);

    if(nodeResult!=NULL && nodeResult->status == LUA_ACT_WAITING)
    {
        if(statusIn)
        {
            nodeResult->status = LUA_ACT_ERROR;
            if(nodeResult->ptrBack)
            {
                nodeResult->ptrBack->status = LUA_ACT_ERROR;
            }
            
        }
        else
        {
            nodeResult->status = LUA_ACT_FINISHED;
            if(nodeResult->ptrBack)
            {
                nodeResult->ptrBack->status = LUA_ACT_FINISHED;
            }            
        }
        
        if(dataIn)
        {
            printf("data recv in msg\n");
            if(nodeResult->ptrBack->expectDataType == EXPECT_INT_DATA)
            {
                memcpy(&nodeResult->ptrBack->un_data.data_i,dataIn,sizeof(int));
            }
            else if(nodeResult->ptrBack->expectDataType == EXPECT_FLOAT_DATA)
            {
                memcpy(&nodeResult->ptrBack->un_data.data_f,dataIn,sizeof(float));
            }
            char * ptrData = (char *)dataIn;
			delete [] ptrData;
        }
        
    }
    //printf("updateCtlNode end\n");

    return 0;
}

pthread_mutex_t mutex_set_status = PTHREAD_MUTEX_INITIALIZER;

act_ctl_node * setExpectData(sock_act_ctl * pstCtlDataIn)
{
    if(pstCtlDataIn==NULL)
    {
        return NULL;
    }
    act_ctl_node * pstRet = NULL;
    act_ctl_node *nodeResult =  findCtlNodeInTable(pstCtlDataIn->actId);
    
    if(nodeResult==NULL)
    {
        sock_act_ctl * pstActCtl = (sock_act_ctl*)malloc(sizeof(sock_act_ctl));
        memcpy(pstActCtl,pstCtlDataIn,sizeof(sock_act_ctl));
        pstRet = registerCtlData(pstActCtl);
    }
    else
    {

        //std::cout<<"set status to waiting, actid="<<pstCtlDataIn->actId<<std::endl;
        pthread_mutex_lock(&mutex_set_status);
        if(nodeResult->status == LUA_ACT_WAITING)
        {
            pstRet = NULL;
        }
        else
        {
            nodeResult->actId = pstCtlDataIn->actId;
            nodeResult->status = pstCtlDataIn->status;
            if(nodeResult->ptrBack==NULL)
            {
            nodeResult->ptrBack =  (sock_act_ctl*)malloc(sizeof(sock_act_ctl));           
            }
            memcpy(nodeResult->ptrBack,pstCtlDataIn,sizeof(sock_act_ctl));
            pstRet = nodeResult;
        }
        pthread_mutex_unlock(&mutex_set_status);
    }

    return pstRet;
}

int msgProc_exitFlag = 0;

ack_msg_node* get_msg_buff_head(void)
{
    ack_msg_node* nodeRet = NULL;
    if(g_list_buff_msg.head)
    {
       nodeRet = g_list_buff_msg.head;
       g_list_buff_msg.head = g_list_buff_msg.head->next;
       if(g_list_buff_msg.head == NULL)
       {
           g_list_buff_msg.tail = NULL;
       }
    }
    return nodeRet;
}

void * msgProcFunc(void * paramIn)
{
    ack_msg_node * msgNode = NULL;

    while(!msgProc_exitFlag)
    {       
       msgNode = get_msg_buff_head();

       if(msgNode)
       {
            //proc msg here.
            printf("in msgProcFunc, node[%x],moduleType:%d,actType:%d,component:%d\n",
            msgNode,
            msgNode->msgHead.moduleType,
            msgNode->msgHead.actType,
            msgNode->msgHead.component);
            int actIdTmp = calcActId(msgNode->msgHead.moduleType,msgNode->msgHead.actType,msgNode->msgHead.component);
            updateCtlNode(actIdTmp,msgNode->msgHead.status,msgNode->paramData);
            delete msgNode;
       }
       else
       {
            usleep(100);
       }

    } 

   pthread_detach(pthread_self());

   return (void*)NULL;
}

void init_msg_buff_list(void)
{
    g_list_buff_msg.head = NULL;
    g_list_buff_msg.tail = NULL;

    pthread_t thread_handl;
    pthread_create(&thread_handl,NULL,msgProcFunc,NULL);
}

void clear_msg_buff_list(void)
{
    ack_msg_node * ptrMsgNode = g_list_buff_msg.head;
    ack_msg_node * ptrMsgNodePrev = ptrMsgNode;
    while(ptrMsgNode)
    {
        if(ptrMsgNode->paramData)
        {
            delete [] (char *)ptrMsgNode->paramData;
            ptrMsgNode->paramData = NULL;
        }
        ptrMsgNodePrev = ptrMsgNode;
        ptrMsgNode = ptrMsgNode->next;
        
        delete ptrMsgNodePrev;
        ptrMsgNodePrev = ptrMsgNode;
    }
}

pthread_mutex_t mutex_msg_mem = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex_msg_data = PTHREAD_MUTEX_INITIALIZER;

int decodeMsgAck(char * dataFromSock,msgAckSock * sockAckMsgOut)
{
    char * ptrDataBuff = dataFromSock;

    memcpy(&sockAckMsgOut->moduleType,ptrDataBuff,sizeof(int));
    ptrDataBuff += sizeof(int);

    memcpy(&sockAckMsgOut->actType,ptrDataBuff,sizeof(int));
    ptrDataBuff += sizeof(int);

    memcpy(&sockAckMsgOut->component,ptrDataBuff,sizeof(int));
    ptrDataBuff += sizeof(int);

    memcpy(&sockAckMsgOut->status,ptrDataBuff,sizeof(int));
    ptrDataBuff += sizeof(int);

    memcpy(&sockAckMsgOut->errorCode,ptrDataBuff,sizeof(int));
    ptrDataBuff += sizeof(int);

    return sizeof(int)*5;
}

int init_msg_buf_node(char * dataFromSock, int sockDataLen,ack_msg_node ** nodeOut )
{
    if(sockDataLen<sizeof(msgAckSock))
    {
        * nodeOut = NULL;
        return 1;
    }
    msgAckSock msgNodeTemp;
    int offset = decodeMsgAck(dataFromSock,&msgNodeTemp);
    // printf("in init_msg_buf_node, moduleType:%d,actType:%d,component:%d,status:%d\n",
    // msgNodeTemp.moduleType,
    // msgNodeTemp.actType,
    // msgNodeTemp.component,
    // msgNodeTemp.status);

    pthread_mutex_lock(&mutex_msg_mem);
    * nodeOut = new ack_msg_node;
    memset(* nodeOut,0,sizeof(ack_msg_node));
    memcpy(&((* nodeOut)->msgHead),&msgNodeTemp,sizeof(msgAckSock));
    pthread_mutex_unlock(&mutex_msg_mem); 

    if(offset < sockDataLen)
    {
        int lenData = sockDataLen - offset;
        pthread_mutex_lock(&mutex_msg_data);
        (* nodeOut)->paramData = new char[lenData];
        memcpy((* nodeOut)->paramData,dataFromSock+offset,lenData);        
        pthread_mutex_unlock(&mutex_msg_data);
    }
    return 0;   
}

void append_msg_buff_node(ack_msg_node * nodeIn)
{
    if(g_list_buff_msg.head == NULL)
    {
       g_list_buff_msg.head = nodeIn;
       g_list_buff_msg.tail = nodeIn;
    }
    else if(g_list_buff_msg.tail)
    {
       g_list_buff_msg.tail->next = nodeIn;
       g_list_buff_msg.tail = g_list_buff_msg.tail->next;
    }
}



int calcActId(int moduleTypeIn,int actTypeIn,int component)
{
    int ret = (moduleTypeIn<<8) + (actTypeIn<<4) + component;

    return ret;
}

int checkActReady(int actId)
{
    act_ctl_node *pstActCtl = findCtlNodeInTable(actId);

    if(pstActCtl==NULL || pstActCtl->status != LUA_ACT_WAITING)
    {
        return 1;
        
    }
    else
    {
        return 0;
    }
}

int cliFd_report = -1;
int serverFd_report = -1;

void * sockServerRecvFunc(void * paramIn)
{
    fd_set rfds;
    struct timeval tv;
    int retval;
    tv.tv_sec = 2;
    tv.tv_usec = 0;
    struct sockaddr_un client_addr;
    socklen_t client_len = 0;
    char recvBuff[256];
    int recvLen = 0;
    while(!sock_exitFlag)
    {
       if(cliFd_report<0)
       {
           if(serverFd_report>0)
           {
               cliFd_report = accept(serverFd_report, (struct sockaddr *)&client_addr, &client_len);
               
           }
           else
           {
               initSockServer(0);
           }
       }
       else
       {
            FD_ZERO(&rfds);
            
            FD_SET(cliFd_report, &rfds);
            tv.tv_sec = 2;
            tv.tv_usec = 0;
            retval = select(cliFd_report+1,&rfds,NULL,NULL,&tv);

            if(retval<0)
            {
                perror("select()");
                close(cliFd_report);
                cliFd_report = -1;
            }
            else if(retval==0)
            {
                //no data available in 2 second;
            }
            else
            {
                //data received;
                if(FD_ISSET(cliFd_report, &rfds))
                {
                    recvLen = 0;
                    //printf("received data in sockRecvFunc,");
                    recvLen = read(cliFd_report,recvBuff,256);

                    //printf("recvLen:%d\n",recvLen);

                    if(recvLen>0)
                    {
                        procReportMsg(recvLen,recvBuff);                  
                    }
                    else
                    {
                        //the romote sock closed.
                        close(cliFd_report);
                        cliFd_report = -1;
                    }
                    
                }
            }
       }
       usleep(100);       
    } 

   pthread_detach(pthread_self());

   return (void*)NULL;
}


int initSockServer(int needCreateThread)
{
    socklen_t client_len = 0;
 
    struct sockaddr_un server_addr;
    struct sockaddr_un client_addr;
 
    // 创建流套接字
    serverFd_report = socket(AF_UNIX, SOCK_STREAM, 0);

    unlink(SOCK_IPC_NOTIFY);
 
    // 设置服务器接收的连接地址和监听的端口
    server_addr.sun_family = AF_UNIX;                   // 指定网络套接字
    snprintf(server_addr.sun_path,sizeof(server_addr.sun_path),"%s",SOCK_IPC_NOTIFY);
 
    // 绑定（命名）套接字
    bind(serverFd_report, (struct sockaddr *)&server_addr, sizeof(sockaddr_un));
 
    // 创建套接字队列，监听套接字
    listen(serverFd_report, 5);
    
    if(needCreateThread)
    {
        pthread_t thread_handle;
        pthread_create(&thread_handle,NULL,sockServerRecvFunc,NULL);
    }   
 
    // 忽略子进程停止或退出信号
    //signal(SIGCHLD, SIG_IGN);
    return 0;
}

int convertLogicTranPos(const void * pcrBoardParamIn, void * logMoveOut)
{

    int ret = -1;
    const MovePcrParam * pstParamIn =  (const MovePcrParam *)pcrBoardParamIn;

    logistic_move * pstLogMoveOut = (logistic_move *)logMoveOut;

    const Position * pstPosSrc = &(pstParamIn->getpos());

    const Position * pstPosDst = &(pstParamIn->putpos());

    if(pstPosSrc->postype() == GPRPC::PosType::PT_DJ1_PCR_TRANS)
    {
        pstLogMoveOut->submodule = SUB_MODULE_LOG1;
        pstLogMoveOut->posTypeInner = LOG1_PCR_BOARD;
        ret = 0;
    }
    else if(pstPosSrc->postype() == GPRPC::PosType::PT_TRANS_CONNECT
    ||pstPosSrc->postype() == GPRPC::PosType::PT_TRANS_CONNECT_BREAK)
    {
        pstLogMoveOut->submodule = SUB_MODULE_LOG2;
        pstLogMoveOut->posTypeInner = LOG2_EXCHANGE12;
        ret = 0;
    }
    else if(pstPosSrc->postype() == GPRPC::PosType::PT_DP8_PCR_TRANS)
    {
        pstLogMoveOut->submodule = SUB_MODULE_LOG2;
        pstLogMoveOut->posTypeInner = LOG2_EXCHANGE_BEFORE;
        ret = 0;
    }
    else 
    {
        return ret;
    }

    pstLogMoveOut++;

    if(pstPosDst->postype() == GPRPC::PosType::PT_TRANS_CONNECT
    ||pstPosDst->postype() == GPRPC::PosType::PT_TRANS_CONNECT_BREAK)
    {
        pstLogMoveOut->submodule = SUB_MODULE_LOG2;
        pstLogMoveOut->posTypeInner = LOG2_EXCHANGE12;
        ret = 0;
    }
    else if(pstPosDst->postype() == GPRPC::PosType::PT_DP8_PCR_TRANS)
    {
        pstLogMoveOut->submodule = SUB_MODULE_LOG2;
        pstLogMoveOut->posTypeInner = LOG2_EXCHANGE_BEFORE;
        ret = 0;
    }
    else if(pstPosDst->postype() == GPRPC::PosType::PT_TRANS_TRAY_3RD
    || pstPosDst->postype() == GPRPC::PosType::PT_TRANS_BASE_3RD)
    {
        pstLogMoveOut->submodule = SUB_MODULE_LOG2;
        pstLogMoveOut->posTypeInner = LOG2_EXCHANGE23;
        ret = 0;
    }
    else 
    {
        return -1;
    }

    return 0; 
}